package com.guchenbo.spark.sql;

import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.visitor.SchemaStatVisitor;
import com.alibaba.druid.stat.TableStat;
import com.alibaba.druid.util.JdbcConstants;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer;
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
import org.apache.spark.sql.execution.SparkSqlParser;
import org.apache.spark.sql.internal.SQLConf;

/**
 * 使用druid的sql解析器，尝试
 *
 * @author guchenbo
 * @date 2022/2/28
 */
public class SqlParserUtils {

    public static String test(String sql) {

        return sql;
    }

    public static List<String> parseTable(String sql) {
        String dbType = JdbcConstants.HIVE;
        return parseTable(sql, dbType);
    }

    public static List<String> parseTable(String sql, String dbType) {
        List<SQLStatement> stmtList = SQLUtils.parseStatements(sql, dbType);
        SchemaStatVisitor statVisitor = SQLUtils.createSchemaStatVisitor(dbType);
        stmtList.get(0).accept(statVisitor);
        return statVisitor.getTables().keySet().stream().map(TableStat.Name::getName).collect(Collectors.toList());
    }

    public static void sparkSqlParse(String sql){
        SparkSqlParser parser = new SparkSqlParser(new SQLConf());

        LogicalPlan logicalPlan = parser.parsePlan(sql);

        LogicalPlan analyzedPlan = SimpleAnalyzer.executeAndCheck(logicalPlan);

        System.out.println(analyzedPlan.analyzed());
        System.out.println(analyzedPlan.toJSON());
    }

    public static void main(String[] args) {
        String sql = "SELECT * FROM gptest.gptest_01_test_01";
        System.out.println(parseTable(sql));

        sparkSqlParse(sql);

        sql = "SELECT DISTINCT m.id id, m.model_id modelId,\n"
                        + "        mi.`name` modelName,\n"
                        + "        mi.version modelVersion,\n"
                        + "        mra.open_flag openFlag,\n"
                        + "        mra.`owner` `owner`,\n"
                        + "        m.template_id templateId,\n"
                        + "        m.report_name reportName,\n"
                        + "        m.detail detail,\n"
                        + "        m.board_id boardId,\n"
                        + "        m.mapping_view mappingView,\n"
                        + "        m.mapping_sql mappingSql,\n"
                        + "        m.monitor_config monitorConfig,\n"
                        + "        m.data_cols dataCols,\n"
                        + "        m.job_id jobId,\n"
                        + "        m.creator creator,\n"
                        + "        m.create_time createTime,\n"
                        + "        m.update_time updateTime,\n"
                        + "        m.state state,\n"
                        + "        m.update_by updateBy,m.delete_flag deleteFlag,m.monitor_type monitorType,m.schedule_flag scheduleFlag,\n"
                        + "        m.execute_config executeConfig,m.variable_config variableConfig,\n"
                        + "        m.period period,m.interval 'interval',m.exec_time execTime,m.cron cron,m.expire_date expireDate,\n"
                        + "        m.policy_name policyName,m.bins_info binsInfo,m.bins_config binsConfig, mi.`code`, mt.template_name templateName\n"
                        + "        FROM monitor m\n"
                        + "        INNER JOIN monitor_report_authority mra\n"
                        + "        ON m.id = mra.data_id AND m.monitor_type = 0 AND mra.type = 'MONITOR_REPORT'\n"
                        + "            AND mra.`owner` = 'chenbo.gu'\n"
                        + "\n"
                        + "\n"
                        + "        INNER JOIN model_info mi\n"
                        + "        ON m.model_id = mi.uuid\n"
                        + "\n"
                        + "\n"
                        + "            LEFT JOIN model_user_rel mur\n"
                        + "            ON mi.code = mur.model_uuid\n"
                        + "\n"
                        + "            LEFT JOIN model_auth_apply maa\n"
                        + "            ON mi.uuid = maa.model_uuid\n"
                        + "\n"
                        + "        INNER JOIN turing.monitor_template mt\n"
                        + "        ON m.template_id = mt.id\n"
                        + "            WHERE mra.open_flag = 1 OR mra.`owner` = 'chenbo.gu' OR\n"
                        + "            (mra.open_flag = 0 AND JSON_EXTRACT(JSON_EXTRACT(authority, '$.personalAuthMap'), CONCAT('$.\"','chenbo.gu', '\"')) IS NOT NULL)\n"
                        + "            OR mi.`owner` = 'chenbo.gu' OR mur.account = 'chenbo.gu' OR (maa.apply_user = 'chenbo.gu' AND auth_status = 1)\n"
                        + "\n"
                        + "        ORDER BY m.update_time DESC";
        System.out.println(parseTable(sql));

        String sqlStr = "with view_alias as ( select `a174`,`a105`,`score`,`a137` as `v1` from (select id_no,type\n"
                        + "         ,cast(info['apply_day'] as string) as apply_day\n"
                        + "         ,cast(info['due_cnt'] as int) as due_cnt\n"
                        + "         ,cast(info['f_1'] as float) as f_1\n"
                        + "         ,cast(info['a137'] as float) as a137\n"
                        + "         ,cast(info['a174'] as string) as a174\n"
                        + "         ,cast(info['a105'] as string) as a105\n"
                        + "         ,cast(info['score'] as float) as score\n"
                        + "         ,cast(info['target'] as int) as target\n"
                        + "         ,cast(info['prob'] as float) as prob\n"
                        + "         ,cast(info['credit_pass'] as int) as credit_pass\n"
                        + "  from (select id_no,type\n"
                        + "              ,str_to_map(concat_ws(',',collect_list(concat_ws(':',feature,cast(value as string))))) as info\n"
                        + "          from turing.model_report_psi_col_df\n"
                        + "         group by id_no,type\n"
                        + "             ) a ) view_alias2 ) \n"
                        + "\n"
                        + "\n"
                        + "select * from (\n"
                        + "\n"
                        + "select v1,v2,v3,x from  ( ( select * from ( select v1,'20' as v2,'30' as v3 from view_alias tablesample(10 rows)    ) Table_1 ) Table_1 )  a \n"
                        + "\n"
                        + "left join  \n"
                        + "\n"
                        + "( ( select * from ( select 'a174' as x,a174 as y,'score' as m ,score as n from view_alias tablesample(10 rows)   ) Table_0 union all select * from ( select\n"
                        + "'a105' as x,a105 as y,'score' as m ,score as n from view_alias tablesample(10 rows)   ) Table_0 ) Table_0 )  b \n"
                        + "\n"
                        + "on a.v1=b.x\n"
                        + "\n"
                        + ") dims\n"
                        + "\n"
                        + ";\n"
                        + "with view_alias as ( select `a105`,`a174`,`score`,`a137` as `v1` from (select id_no,type\n"
                        + "         ,cast(info['apply_day'] as string) as apply_day\n"
                        + "         ,cast(info['due_cnt'] as int) as due_cnt\n"
                        + "         ,cast(info['f_1'] as float) as f_1\n"
                        + "         ,cast(info['a137'] as float) as a137\n"
                        + "         ,cast(info['a174'] as string) as a174\n"
                        + "         ,cast(info['a105'] as string) as a105\n"
                        + "         ,cast(info['score'] as float) as score\n"
                        + "         ,cast(info['target'] as int) as target \n"
                        + "         ,cast(info['prob'] as float) as prob\n"
                        + "         ,cast(info['credit_pass'] as int) as credit_pass          \n"
                        + "  from (select id_no,type\n"
                        + "              ,str_to_map(concat_ws(',',collect_list(concat_ws(':',feature,cast(value as string))))) as info\n"
                        + "          from turing.model_report_psi_col_df\n"
                        + "         group by id_no,type\n"
                        + "             ) a ) view_alias2 ) \n"
                        + "\n"
                        + "\n"
                        + "insert overwrite table turing_monitor.monitor_10748_408_Table_2 \n"
                        + "select * from (\n"
                        + "\n"
                        + "select v1,v2,v3,x from  ( ( select * from ( select v1,'' as v2,'' as v3 from view_alias   ) Table_1 ) Table_1 )  a \n"
                        + "\n"
                        + "left join  \n"
                        + "\n"
                        + "( ( select * from ( select 'a105' as x,a105 as y,'score' as m ,score as n from view_alias  ) Table_0 union all select * from ( select 'a174' as x,a174 as y,'score' as m ,score as n from view_alias  ) Table_0 ) Table_0 )  b \n"
                        + "\n"
                        + "on a.v1=b.x\n"
                        + "\n"
                        + ") dims\n"
                        + "\n"
                        + ";\n"
                        + "with view_alias as ( select `a174`,`a105`,`score`,`a137` as `v1` from (select id_no,type\n"
                        + "         ,cast(info['apply_day'] as string) as apply_day\n"
                        + "         ,cast(info['due_cnt'] as int) as due_cnt\n"
                        + "         ,cast(info['f_1'] as float) as f_1\n"
                        + "         ,cast(info['a137'] as float) as a137\n"
                        + "         ,cast(info['a174'] as string) as a174\n"
                        + "         ,cast(info['a105'] as string) as a105\n"
                        + "         ,cast(info['score'] as float) as score\n"
                        + "         ,cast(info['target'] as int) as target \n"
                        + "         ,cast(info['prob'] as float) as prob\n"
                        + "         ,cast(info['credit_pass'] as int) as credit_pass          \n"
                        + "  from (select id_no,type\n"
                        + "              ,str_to_map(concat_ws(',',collect_list(concat_ws(':',feature,cast(value as string))))) as info\n"
                        + "          from turing.model_report_psi_col_df\n"
                        + "         group by id_no,type\n"
                        + "             ) a ) view_alias2 ) \n"
                        + "\n"
                        + "insert overwrite table turing_monitor.template_408_table_2 \n"
                        + "select * from (\n"
                        + "\n"
                        + "select v1,v2,v3,x from  ( ( select * from ( select v1,'20' as v2,'30' as v3 from view_alias   ) Table_1 ) Table_1 )  a \n"
                        + "\n"
                        + "left join  \n"
                        + "\n"
                        + "( ( select * from ( select 'a174' as x,a174 as y,'score' as m ,score as n from view_alias  ) Table_0 union all select * from ( select 'a105' as x,a105 as y,'score' as m ,score as n from view_alias  ) Table_0 ) Table_0 )  b \n"
                        + "\n"
                        + "on a.v1=b.x\n"
                        + "\n"
                        + ") dims\n"
                        + "\n"
                        + "\n"
                        + "\n"
                        + "\n"
                        + "\n"
                        + "\n"
                        + "\n"
                        + "\n";

         sql = "SELECT * FROM gptest.gptest_01_test_01";
        sql = "select id_no,type\n"
            + "         ,cast(info['apply_day'] as string) as apply_day\n"
            + "         ,cast(info['due_cnt'] as int) as due_cnt\n"
            + "         ,cast(info['f_1'] as float) as f_1\n"
            + "         ,cast(info['a137'] as float) as a137\n"
            + "         ,cast(info['a174'] as string) as a174\n"
            + "         ,cast(info['a105'] as string) as a105\n"
            + "         ,cast(info['score'] as float) as score\n"
            + "         ,cast(info['target'] as int) as target \n"
            + "         ,cast(info['prob'] as float) as prob\n"
            + "         ,cast(info['credit_pass'] as int) as credit_pass          \n"
            + "  from (select id_no,type\n"
            + "              ,str_to_map(concat_ws(',',collect_list(concat_ws(':',feature,cast(value as string))))) as info\n"
            + "          from turing.model_report_psi_col_df\n"
            + "         group by id_no,type\n"
            + "             ) a";

        for (String s : sqlStr.split(";")) {
            System.out.println(s);
            System.out.println(parseTable(s.trim()));
        }
    }
}
